package com.atguigu.realtime.clients;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.realtime.constants.TopicConstant;
import com.atguigu.realtime.utils.KafkaProducerUtil;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * Created by Smexy on 2022/8/27
 *
 * ①先创建一个客户端对象CanalConnector
 * ②使用客户端对象连接 Canal server端
 * ③订阅表,并拉取数据
 * ④解析订阅到的数据
 * ⑤将数据写入kafka
 */
public class OrderInfoClient
{
    public static void main(String[] args) throws InvalidProtocolBufferException {

        /*
            SocketAddress address：  canalsever的主机名和端口号。
                                                参考: canal.properties
                                                    canal.ip = hadoop103
                                                    canal.port = 11111
             String destination：    参考: canal.properties
                                            指instance.properties文件所在的目录名
                                            canal.destinations = example
              String username:   目前1.1.2不支持
              String password:  目前1.1.2不支持
         */
        //①先创建一个客户端对象CanalConnector
        CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop103", 11111), "example", null
            , null);

        //②使用客户端对象连接 Canal server端
        canalConnector.connect();

        // ③订阅表 目前GMV需求，只需要采集  orderinfo表中每一笔新增的订单的 total_amount
        // 格式： 库名.表名
        canalConnector.subscribe("220409.order_info");


        //拉取数据
        while(true){

            Message message = canalConnector.get(100);

            if (message.getId() == -1){

                //当前没有拉取到数据
                System.out.println("当前没有新数据了，歇5s....");
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                continue;

            }

            //说明拉到数据，解析数据
            //System.out.println(message);
            List<CanalEntry.Entry> entries = message.getEntries();

            for (CanalEntry.Entry entry : entries) {

                //sql写操作的表名
                String tableName = entry.getHeader().getTableName();

                //写入的数据，需要反序列化再使用
                ByteString storeValue = entry.getStoreValue();

                // 如果你只处理 insert,update,delete，它们的 EntryType 都是 ROWDATA
                CanalEntry.EntryType entryType = entry.getEntryType();

                if (entryType.equals(CanalEntry.EntryType.ROWDATA)){
                    parseData(storeValue);
                }


            }


        }



    }

    // 当前只奥 orderInfo这个表的 insert操作
    private static void parseData(ByteString storeValue) throws InvalidProtocolBufferException {

        // ByteString storeValue: 反序列化之前的 写入的数据，不能直接使用
        //  CanalEntry.RowChange rowChange: 反序列化之后的 写入的数据，能直接使用。代表一个SQL的写操作(有可能会影响多行)变化。
        //反序列化的固定套路
        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);

        //EventType : 写操作的关键字   insert ,update等
        if (rowChange.getEventType().equals(CanalEntry.EventType.INSERT)){

            //这条insert语句，造成的多行的写操作变化
            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();

            //一个rowData写入的一行
            for (CanalEntry.RowData rowData : rowDatasList) {

                //把一行封装为一个JSON
                JSONObject jsonObject = new JSONObject();

                // BeforeColumnsList(): 写入前，每一列的信息。   update
                // AfterColumnsList(): 写入后，每一列的信息。    insert,update
                List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();

                for (CanalEntry.Column column : afterColumnsList) {

                   jsonObject.put( column.getName(),column.getValue());

                }

                System.out.println(jsonObject);

                //发到kafka
                KafkaProducerUtil.sendData(jsonObject.toString(), TopicConstant.ORDER_INFO);


            }
        }

    }
}
